package com.corn.flink.lesson4;

import com.corn.flink.lesson3.Person;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * watermark demo
 *
 * @author JimWu
 * @date 2023/3/1 16:55
 **/
public class FlinkWatermarkMonotonousTimestampsDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Event> ds = env.fromElements(
                new Event("add", 1000),
                new Event("delete", 2000),
                new Event("update", 3000),
                new Event("select", 5000),
                new Event("add", 6000),
                new Event("delete", 7000),
                new Event("select", 8000)
        );

        // 设置水位线策略 使用flink 内置的水位线MonotonousTimestamps单调递增的水位线 适用于有序数据
        ds.assignTimestampsAndWatermarks(
                WatermarkStrategy.<Event>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event element, long recordTimestamp) {
                                // 此时也可以获取Person对象中带有时间戳定义
                                return element.timestamp;
                            }
                        })
        );


        ds.print();
        env.execute();
    }
}
